package basicutil.pipeline.core;

import basicutil.common.ThreadUtil;
import basicutil.pipeline.DataMessage;
import basicutil.pipeline.DataType;
import basicutil.pipeline.Processor;
import basicutil.pipeline.State;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 数据处理节点 - 核心类
 */
public class ProcessorStage {
    private static final Logger log = LoggerFactory.getLogger(ProcessorStage.class);

    private final String id;
    private Processor processor; // 处理逻辑



    private final List<InputPort> inputPorts;
    private final List<OutputPort> outputPorts;

    private final AtomicBoolean running;
    // 配置参数
    private AtomicLong processingInterval; // 处理间隔(ms)

    private final AtomicLong processedCount;


    private Thread waitingThread;
    private State state;
    private String waitObj;


    public ProcessorStage(String id) {
        this.id = id;
        this.inputPorts = new ArrayList<>();
        this.outputPorts = new ArrayList<>();

        this.running = new AtomicBoolean(false);

        this.processedCount = new AtomicLong(0);

        // 默认配置
        this.processingInterval = new AtomicLong(1000); // 1秒

    }

    // 端口管理方法
    public void addInputPort(DataType dataType) {
        String portId = getId() + "-in-" + (inputPorts.size());
        InputPort port = new InputPort(portId, dataType);
        inputPorts.add(port);
    }

    public void addInputPort(InputPort port) {
        inputPorts.add(port);
    }

    public void addOutputPort(DataType dataType) {
        String portId = getId() + "-out-" + (outputPorts.size());
        OutputPort port = new OutputPort(portId, dataType);
        outputPorts.add(port);

    }

    public void addOutputPort(OutputPort port) {
        outputPorts.add(port);

    }

    public InputPort getInputPort(int portId) {
        return inputPorts.get(portId);
    }


    public OutputPort getOutputPort(int portId) {
        return outputPorts.get(portId);
    }

    // 配置方法
    public void setProcessingInterval(long intervalMs) {
        this.processingInterval.set(intervalMs);

    }


    public void setProcessor(Processor processor) {
        this.processor = processor;
    }

    // 生命周期控制
    public void start() {
        if (running.compareAndSet(false, true)) {
            ThreadUtil.runLaterSafely(() -> {

                while (isRunning()) {
                    runOnce();

                    waitingThread = Thread.currentThread();
                    state = State.WAIT_INTERVAL;
                    waitObj = "interval";

                    ThreadUtil.sleep(processingInterval.get());
                    state = State.RUNNING;
                }
            });
        }
    }

    public void startOnce() {
        if (running.compareAndSet(false, true)) {
            runOnce();
            stop();
        }

    }

    public void stop() {
        if (running.compareAndSet(true, false)) {
            if (waitingThread != null && (state == State.WAIT_PARAM || state == State.WAIT_INTERVAL)) {
                waitingThread.interrupt();
            }
            state = State.STOP;
        }
    }


    void runOnce() {
        try {
            List<DataMessage> results;
            ArrayList<DataMessage> paramList = new ArrayList<>();
            if (!inputPorts.isEmpty()) {
                for (InputPort inputPort : inputPorts) {
                    if (!isRunning()) {
                        stop();
                        return;
                    }
                    waitingThread = Thread.currentThread();
                    state = State.WAIT_PARAM;
                    waitObj = inputPort.getId();
                    DataMessage dataMessage = inputPort.waitMessage();
                    state = State.RUNNING;
                    paramList.add(dataMessage);
                }
            }
            if (!isRunning()) {
                stop();
                return;
            }
            results = processor.process(paramList, this);
            if (results != null && !results.isEmpty()) {
                for (int i = 0; i < results.size(); i++) {
                    if (outputPorts.size() > i) {
                        DataMessage message = results.get(i);
                        outputPorts.get(i).receiveMessage(message);
                    }
                }
            }

            processedCount.addAndGet(1);
            return;
        } catch (InterruptedException e) {
            return;
        } catch (Exception e) {
            e.printStackTrace();
            running.set(false);
            return;
        }

    }

    public long getProcessedCount() {
        return processedCount.get();
    }

    public boolean isRunning() {
        return running.get();
    }


    public String getId() {
        return id;
    }

    public Thread getWaitingThread() {
        return waitingThread;
    }

    public State getState() {
        return state;
    }

    public String getWaitObj() {
        return waitObj;
    }
    public List<InputPort> getInputPorts() {
        return inputPorts;
    }

    public List<OutputPort> getOutputPorts() {
        return outputPorts;
    }
    @Override
    protected void finalize() throws Throwable {
        stop();
        super.finalize();
    }
}
